package com.xiashitech.apiservice.grpc;

import cn.hutool.json.JSONUtil;
import com.xiashitech.apiservice.Constants.ErrorCode;
import com.xiashitech.apiservice.service.IAgentSystemStatusService;
import com.xiashitech.interfaces.agent.constants.AgentStatus;
import com.xiashitech.interfaces.agent.dto.SystemInfoDTO;
import com.xiashitech.interfaces.apiservice.dto.system.SystemInfoRequest;
import com.xiashitech.interfaces.apiservice.dto.system.SystemInfoResponse;
import com.xiashitech.interfaces.grpc.RPCAgentStatusServiceGrpc;
import com.xiashitech.interfaces.grpc.RPCSystemInfoRequest;
import com.xiashitech.interfaces.grpc.RPCSystemInfoResponse;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import net.devh.boot.grpc.server.service.GrpcService;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
//@Service
@GrpcService
public class AgentStatusService extends RPCAgentStatusServiceGrpc.RPCAgentStatusServiceImplBase {

    @Autowired
    private IAgentSystemStatusService agentSystemStatusService;
    // 一个 Agent 维持一个，采用 Hash + List 结构，是为了支持多个 agent 配置相同的 systemId + serviceId 的场景
    private volatile ConcurrentHashMap<String, List<StreamObserver<RPCSystemInfoResponse>>> responseSystemInfoMap = new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);

    private AgentStatusService() {
        scheduledExecutorService.scheduleWithFixedDelay(()->{
            // 检查链接是否有效，链接有效，表示 agent 还存活
            try {
                synchronized (responseSystemInfoMap) {
                    Iterator<Map.Entry<String, List<StreamObserver<RPCSystemInfoResponse>>>> mapIterator = responseSystemInfoMap.entrySet().iterator();
                    while (mapIterator.hasNext()) {
                        Map.Entry<String, List<StreamObserver<RPCSystemInfoResponse>>> entry = mapIterator.next();
                        String key = entry.getKey();
                        String [] keyArr = key.split("_");
                        List<StreamObserver<RPCSystemInfoResponse>> soList = responseSystemInfoMap.get(key);
                        if (soList == null || soList.size() == 0) {
                            log.info(key + " 目标 agent 可能已下线，没有推送通道(A)");
                            // 暂时保留 mapIterator.remove();
                            // 不更新状态，由原先定时任务根据心跳判断
                            continue;
                        }

                        log.info(key + ", agentStatus grpc client size: " + soList.size() + ", mapsize: " + responseSystemInfoMap.size());
                        int idx = 0;
                        Iterator<StreamObserver<RPCSystemInfoResponse>> iterator = soList.iterator();
                        while (iterator.hasNext()) {
                            ServerCallStreamObserver scso = (ServerCallStreamObserver) iterator.next();
                            if (scso.isCancelled()) {
                                log.info(key + ", " + idx + " grpc client is offline, remove the SI");
                                iterator.remove(); // 需要用迭代器的 remove，不能用列表的 remove
                                // 不标记 agent 下线，由由原先定时任务根据心跳判断
                                continue;
                            }
                            idx++;
                        }
                        // 如果多个 agent 配置了相同的 systemId 和 serviceId，只要有一个存活，就认为还存活，更新心跳信息
                        if(idx > 0)
                            agentSystemStatusService.updateAgentHeartBeatByCache(keyArr[0],keyArr[1]);
                    }
                }
            } catch (Throwable t) {
                t.printStackTrace();
            }
        }, 30, 5, TimeUnit.SECONDS);
    }
    @Override
    public StreamObserver<RPCSystemInfoRequest> biAgentStatus(StreamObserver<RPCSystemInfoResponse> responseObserver) {
        log.info("biAgentStatus enter, agentSystemStatusService is " + agentSystemStatusService);
        StreamObserver<RPCSystemInfoRequest> requestObserver = new StreamObserver<RPCSystemInfoRequest>() {
            @Override
            public void onNext(RPCSystemInfoRequest value) {
                String reqstr = value.getSystemInfoRequestJson();
                if("AGENTHEARTBEAT".equals(reqstr)) {
                    log.info("received the AgentStatus heartbeat request, response AGENTHEARTBEAT");
                    RPCSystemInfoResponse rpcSystemInfoResponse = RPCSystemInfoResponse.newBuilder()
                            .setSystemInfoResponseJson("AGENTHEARTBEAT")
                            .build();
                    responseObserver.onNext(rpcSystemInfoResponse);
                    return;
                }
                if(reqstr == null || "".equals(reqstr)) { // 表示是验证链接的请求，给出响应，可用于检查 channel，当前没用
                    log.info("received the check blocking request");
                    synchronized (responseObserver) { // 可能并发访问，并发度不高，应该是单线程，长跨度访问
                        responseObserver.onNext(buildEmptyResponse("", ""));
                    }
                    return;
                }

                SystemInfoRequest systemInfoRequest = JSONUtil.toBean(reqstr, SystemInfoRequest.class);
                String systemId = systemInfoRequest.getId();
                String micrroServiceId = systemInfoRequest.getMicroServiceId();
                String key = systemId+"_"+micrroServiceId;
                synchronized (responseSystemInfoMap) {
                    List<StreamObserver<RPCSystemInfoResponse>> streamObservers = responseSystemInfoMap.get(key);
                    if (streamObservers == null) {
                        streamObservers = new ArrayList<>();
                        streamObservers.add(responseObserver);
                        responseSystemInfoMap.put(key, streamObservers);
                    } else {
                        // 检查一下 responseObserver 是否发生改变
                        boolean isExist = false; // 因为接口会被调用多次，同一个流下 onNext 也会被调用多次，相同的情况很正常
                        for (StreamObserver<RPCSystemInfoResponse> so : streamObservers) {
                            if (so == responseObserver) {
                                isExist = Boolean.TRUE;
                                log.info("exist the same response observer, NORMAL");
                            }
                        }
                        if (!isExist) {
                            streamObservers.add(responseObserver);
                        }
                    }
                }

                if(agentSystemStatusService != null) {
                    log.info(Thread.currentThread().getId() + ", agentSystemStatusService NOT NULL");
                    SystemInfoDTO systemInfoDTO = agentSystemStatusService.selectAgentStatusByCache(systemInfoRequest.getId(), systemInfoRequest.getMicroServiceId());
                    agentSystemStatusService.updateAgentHeartBeatByCache(systemInfoRequest.getId(), systemInfoRequest.getMicroServiceId());
                    //  {"id":"yongdavip","microServiceId":"vipstruts","agentStatus":1,"agentStatusCreate":1687944197000,"agentStatusUpdate":1690610492000,"agentStatusHeartBeat":1690610606000}
                    log.info(Thread.currentThread().getId() + ", systemInfoDTO: " + JSONUtil.toJsonStr(systemInfoDTO));
                    List<SystemInfoDTO> systemInfos = new ArrayList<>();
                    systemInfos.add(systemInfoDTO);
                    SystemInfoResponse systemInfoResponse = SystemInfoResponse.builder()
                            .code(ErrorCode.RESULT_1001.getCode())
                            .message(ErrorCode.RESULT_1001.getMessage())
                            .systemInfos(systemInfos)
                            .build();
                    RPCSystemInfoResponse rpcSystemInfoResponse = RPCSystemInfoResponse.newBuilder()
                            .setSystemInfoResponseJson(JSONUtil.toJsonStr(systemInfoResponse))
                            .build();
                    log.info(Thread.currentThread().getId() + ", RPCSystemInfoResponse: " + rpcSystemInfoResponse);
                    synchronized (responseObserver) {
                        responseObserver.onNext(rpcSystemInfoResponse);
                    }
                } else {
                    log.info(Thread.currentThread().getId() + ", agentSystemStatusService is NULL, EXCEPTION!!!");
                    RPCSystemInfoResponse rpcSystemInfoResponse = buildEmptyResponse(systemInfoRequest.getId(), systemInfoRequest.getMicroServiceId());
                    synchronized (responseObserver) {
                        responseObserver.onNext(rpcSystemInfoResponse);
                    }
                }

                // 不能 onCompleted，调用了这方法，表示服务端已经发送完成，不能后续再发数据
                // responseObserver.onCompleted();
                log.info(Thread.currentThread().getId() + ", end agentStatus run");
            }

            @Override
            public void onError(Throwable t) {
                log.info("biAgentStatus onError" + t.getMessage());
                if (responseObserver instanceof ServerCallStreamObserver) {
                    ServerCallStreamObserver serverCallStreamObserver = ((ServerCallStreamObserver) responseObserver);
                    if (serverCallStreamObserver.isCancelled()) {
                        //client close the stream.
                    } else {
                        try {
                            serverCallStreamObserver.onCompleted();
                        } catch (Throwable throwable) {
                            //ignore
                        }
                    }
                }
            }

            @Override
            public void onCompleted() {
                log.info("biAgentStatus onCompleted");
                if (responseObserver instanceof ServerCallStreamObserver) {
                    ServerCallStreamObserver serverCallStreamObserver = ((ServerCallStreamObserver) responseObserver);
                    if (serverCallStreamObserver.isCancelled()) {
                        //client close the stream.
                    } else {
                        try {
                            serverCallStreamObserver.onCompleted();
                        } catch (Throwable throwable) {
                            //ignore
                        }
                    }
                }
            }
        };

        return requestObserver;
    }
    private RPCSystemInfoResponse buildEmptyResponse(String systemId, String microSystemId) {
        SystemInfoDTO systemInfoDTO = SystemInfoDTO.builder()
                .systemExist(false)
                .agentStatus(0)
                .id(systemId)
                .name(microSystemId)
                .agentStatusCreate(new Date()) //LocalDateTimeUtil.format(LocalDateTime.now(), "yyyy-MM-dd HH:mm:ss"))
                .agentStatusUpdate(new Date()) //String.valueOf(new DateTime())) // 这个格式是合适的，yyyy-MM-dd HH:mm:ss
                .agentStatusHeartBeat(new Date()) //String.valueOf(random.nextInt(1000)))
                .build();
        List<SystemInfoDTO> systemInfos = new ArrayList<>();
        systemInfos.add(systemInfoDTO);
        SystemInfoResponse systemInfoResponse = SystemInfoResponse.builder()
                .code(ErrorCode.RESULT_1001.getCode())
                .message(ErrorCode.RESULT_1001.getMessage())
                .systemInfos(systemInfos)
                .build();
        RPCSystemInfoResponse rpcSystemInfoResponse = RPCSystemInfoResponse.newBuilder()
                .setSystemInfoResponseJson(JSONUtil.toJsonStr(systemInfoResponse))
                .build();
        return rpcSystemInfoResponse;
    }

    @Override
    public void grpcRegistAgentStatus(RPCSystemInfoRequest request, StreamObserver<RPCSystemInfoResponse> responseObserver) {
        log.info("grpcRegistAgentStatus enter, agentSystemStatusService is " + agentSystemStatusService);

        String siStr = request.getSystemInfoRequestJson();
        log.info(Thread.currentThread().getId() + ", grpcSyncAgentStatus Enterd， request string: " + siStr);
        if(siStr == null || "".equals(siStr)) {
            SystemInfoResponse systemInfoResponse = SystemInfoResponse.builder().code(ErrorCode.RESULT_4014.getCode()).message(ErrorCode.RESULT_4014.getMessage()).build();
            RPCSystemInfoResponse rpcSystemInfoResponse = RPCSystemInfoResponse.newBuilder().setSystemInfoResponseJson(JSONUtil.toJsonStr(systemInfoResponse)).build();
            responseObserver.onNext(rpcSystemInfoResponse);
            responseObserver.onCompleted();
            return;
        }

        log.info(Thread.currentThread().getId() + ", grpcSyncAgentStatus after success get agentSystemStatusService");
        SystemInfoRequest systemInfoRequest = JSONUtil.toBean(siStr, SystemInfoRequest.class);
        SystemInfoDTO systemInfoDTO = agentSystemStatusService.selectAgentStatusByCache(systemInfoRequest.getId(),systemInfoRequest.getMicroServiceId());
        log.info(Thread.currentThread().getId() + ", grpcSyncAgentStatus after success get SystemInfoDTO: " + systemInfoDTO);
        if(systemInfoRequest.getAgentStatus() == AgentStatus.NOT_RUN.getValue()) {
            if(systemInfoDTO != null && systemInfoDTO.getAgentStatus() != null && systemInfoDTO.getAgentStatus() == AgentStatus.NOT_RUN.getValue()) {
                SystemInfoResponse systemInfoResponse = SystemInfoResponse.builder().code(ErrorCode.RESULT_1002.getCode()).message(ErrorCode.RESULT_1002.getMessage()).build();
                RPCSystemInfoResponse rpcSystemInfoResponse = RPCSystemInfoResponse.newBuilder().setSystemInfoResponseJson(JSONUtil.toJsonStr(systemInfoResponse)).build();
                responseObserver.onNext(rpcSystemInfoResponse);
                responseObserver.onCompleted();
                return ;
            }
        }
        if(systemInfoDTO == null) {
            log.info(Thread.currentThread().getId() + ", grpcSyncAgentStatus create new agent: ");
            agentSystemStatusService.insertAgentStatus(SystemInfoDTO.builder().id(systemInfoRequest.getId()).microServiceId(systemInfoRequest.getMicroServiceId()).agentStatus(systemInfoRequest.getAgentStatus()).build());
        } else if (systemInfoDTO != null) {
            if(systemInfoDTO.getAgentStatus() != null && systemInfoDTO.getAgentStatus() == AgentStatus.BY_PASS.getValue()) {
                SystemInfoResponse systemInfoResponse = SystemInfoResponse.builder().code(ErrorCode.RESULT_1002.getCode()).message(ErrorCode.RESULT_1002.getMessage()).build();
                RPCSystemInfoResponse rpcSystemInfoResponse = RPCSystemInfoResponse.newBuilder().setSystemInfoResponseJson(JSONUtil.toJsonStr(systemInfoResponse)).build();
                responseObserver.onNext(rpcSystemInfoResponse);
                responseObserver.onCompleted();
                return ;
            }
            log.info(Thread.currentThread().getId() + ", grpcSyncAgentStatus update agent infor");
            agentSystemStatusService.updateAgentStatusByCache(systemInfoRequest.getId(), systemInfoRequest.getMicroServiceId(), systemInfoRequest.getAgentStatus(), Boolean.TRUE);
        }

        SystemInfoResponse systemInfoResponse = SystemInfoResponse.builder().code(ErrorCode.RESULT_1001.getCode()).message(ErrorCode.RESULT_1001.getMessage()).build();
        RPCSystemInfoResponse rpcSystemInfoResponse = RPCSystemInfoResponse.newBuilder().setSystemInfoResponseJson(JSONUtil.toJsonStr(systemInfoResponse)).build();
        responseObserver.onNext(rpcSystemInfoResponse);
        responseObserver.onCompleted();
        log.info(Thread.currentThread().getId() + ", grpcSyncAgentStatus sucess finished");
    }


    public void pushAgentStatusToAgent(SystemInfoResponse systemInfoResponse) {
        // 获取代表对应的 Agent 的 StreamObserver
        List<SystemInfoDTO> siList = systemInfoResponse.getSystemInfos();
        if(siList == null || siList.size() == 0) {
            log.info("WARN: 没找到要推送的系统信息");
            return;
        }
        log.info("pushAgentStatusToAgent enter begin ");
        for(SystemInfoDTO sidto : siList) {
            String systemId = sidto.getId();
            String microServiceId = sidto.getMicroServiceId();
            String key = systemId + "_" + microServiceId;
            // 为避免参数中带有多个不同的 systemId 和 serviceId（通常不会），在这循环中重新构建 SystemInfoResponse
            List<SystemInfoDTO> systemInfos = new ArrayList<>();
            systemInfos.add(sidto);
            SystemInfoResponse singleSystemInfoResponse = SystemInfoResponse.builder()
                    .code(ErrorCode.RESULT_1001.getCode())
                    .message(ErrorCode.RESULT_1001.getMessage())
                    .systemInfos(systemInfos)
                    .build();
            RPCSystemInfoResponse rpcSystemInfoResponse = RPCSystemInfoResponse.newBuilder()
                    .setSystemInfoResponseJson(JSONUtil.toJsonStr(singleSystemInfoResponse))
                    .build();

            log.info("pushAgentStatusToAgent begin key: " + key);

            synchronized (responseSystemInfoMap) {
                List<StreamObserver<RPCSystemInfoResponse>> soList = responseSystemInfoMap.get(key);
                if (soList == null || soList.size() == 0) {
                    log.info(key + " 目标 agent 可能已下线，没有推送通道(B)");
                    continue;
                }

                Iterator<StreamObserver<RPCSystemInfoResponse>> iterator = soList.iterator();
                while (iterator.hasNext()) {
                    ServerCallStreamObserver scso = (ServerCallStreamObserver) iterator.next();
                    if(!scso.isCancelled()) { // 不做删除，由定时任务统一完成
                        synchronized (scso) { // 锁套锁，但这里应该不会出现死锁
                            log.info("pushAgentStatusToAgent pushed: " + sidto);
                            scso.onNext(rpcSystemInfoResponse);
                        }
                    }
                }
            }
        }
    }

    public void grpcSyncAgentHost(RPCSystemInfoRequest request, StreamObserver<RPCSystemInfoResponse> responseObserver) {
        String siStr = request.getSystemInfoRequestJson();
        if(siStr == null || "".equals( siStr)) {
            SystemInfoResponse systemInfoResponse = SystemInfoResponse.builder().code(ErrorCode.RESULT_4014.getCode()).message(ErrorCode.RESULT_4014.getMessage()).build();
            RPCSystemInfoResponse rpcSystemInfoResponse = RPCSystemInfoResponse.newBuilder().setSystemInfoResponseJson(JSONUtil.toJsonStr(systemInfoResponse)).build();
            responseObserver.onNext(rpcSystemInfoResponse);
            responseObserver.onCompleted();
            return;
        }

        SystemInfoRequest systemInfoRequest = JSONUtil.toBean(siStr, SystemInfoRequest.class);
        SystemInfoDTO systemInfoDTO = agentSystemStatusService.selectAgentStatusByCache(systemInfoRequest.getId(),systemInfoRequest.getMicroServiceId());
        if(systemInfoDTO == null) {
            SystemInfoResponse systemInfoResponse = SystemInfoResponse.builder().code(ErrorCode.RESULT_4016.getCode()).message(ErrorCode.RESULT_4016.getMessage()).build();
            RPCSystemInfoResponse rpcSystemInfoResponse = RPCSystemInfoResponse.newBuilder().setSystemInfoResponseJson(JSONUtil.toJsonStr(systemInfoResponse)).build();
            responseObserver.onNext(rpcSystemInfoResponse);
            responseObserver.onCompleted();
            return;
        } else if (systemInfoDTO != null) {
            if(systemInfoDTO.getHost() != null && systemInfoDTO.getHost().equals(systemInfoRequest.getHost())) {
                SystemInfoResponse systemInfoResponse = SystemInfoResponse.builder().code(ErrorCode.RESULT_1002.getCode()).message(ErrorCode.RESULT_1002.getMessage()).build();
                RPCSystemInfoResponse rpcSystemInfoResponse = RPCSystemInfoResponse.newBuilder().setSystemInfoResponseJson(JSONUtil.toJsonStr(systemInfoResponse)).build();
                responseObserver.onNext(rpcSystemInfoResponse);
                responseObserver.onCompleted();
                return ;
            }
            agentSystemStatusService.updateAgentHostByCache(systemInfoRequest.getId(), systemInfoRequest.getMicroServiceId(), systemInfoRequest.getHost(), Boolean.FALSE);
        }
        SystemInfoResponse systemInfoResponse = SystemInfoResponse.builder().code(ErrorCode.RESULT_1001.getCode()).message(ErrorCode.RESULT_1001.getMessage()).build();
        RPCSystemInfoResponse rpcSystemInfoResponse = RPCSystemInfoResponse.newBuilder().setSystemInfoResponseJson(JSONUtil.toJsonStr(systemInfoResponse)).build();
        responseObserver.onNext(rpcSystemInfoResponse);
        responseObserver.onCompleted();
    }
}
